package cn.doitedu.demo2;

import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class RuleCalculator_02 implements RuleCalculator{

    private String ruleId = "rule-002";

    ValueState<Integer> ACNT;
    ValueState<Integer> IDX;
    ValueState<Integer> SEQ_CNT;
    final String[] eventSeq = {"Q","C","Y"};
    JSONObject message;
    Table table;


    @Override
    public void init(RuntimeContext runtimeContext) throws IOException {

        ACNT = runtimeContext.getState(new ValueStateDescriptor<Integer>("A_CNT", Integer.class));
        IDX = runtimeContext.getState(new ValueStateDescriptor<Integer>("IDX", Integer.class));
        SEQ_CNT = runtimeContext.getState(new ValueStateDescriptor<Integer>("SEQ_CNT", Integer.class));

        message = new JSONObject();
        message.put("rule_id","rule-002");

        // 创建hbase用户画像数据查询的客户端
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        // hbase-site.xml
        conf.set("hbase.zookeeper.quorum","doitedu:2181");

        Connection connection = ConnectionFactory.createConnection(conf);
        table = connection.getTable(TableName.valueOf("user_profile_tags"));


    }

    @Override
    public void calc(UserAction userAction, Collector<String> out) throws IOException {
        String eventId = userAction.getEvent_id();

        /* *
         * 用户发生 A行为：3次以上
         * 用户发生  [ Q C Y ] 事件序列:  2次以上
         * 当他做 C行为时，触发规则
         */

        // A行为统计
        if(eventId.equals("A")){
            int cnt = ACNT.value() == null? 1: ACNT.value()+1;
            ACNT.update(cnt);
        }

        // [ Q C Y ] 事件序列的发生次数统计
        int idx = IDX.value() == null ? 0: IDX.value();
        if(eventId.equals( eventSeq[idx]  )){
            // 如果是，则要把idx 递增
            int newIdx = idx+1;
            // 如果idx递增后等于3了，说明这个序列到此被完整发生过了
            if(newIdx == 3){
                // 重置idx
                newIdx = 0;
                // 增加完整发生次数
                SEQ_CNT.update( SEQ_CNT.value() == null? 1 : SEQ_CNT.value()+1  );
            }
            IDX.update(newIdx);
        }

        // 触发事件C行为 处理
        if(eventId.equals("C")){


            Integer aCount = ACNT.value();
            Integer seqCount = SEQ_CNT.value();

            // 判断该用户的动态画像统计条件是否已经满足
            if(aCount != null && seqCount != null && aCount>=3 && seqCount>=2){

                // 查询该用户的静态画像标签
                Get get = new Get(Bytes.toBytes(userAction.getUser_id()));
                get.addColumn(Bytes.toBytes("f"),Bytes.toBytes("age"));
                get.addColumn(Bytes.toBytes("f"),Bytes.toBytes("gender"));
                get.addColumn(Bytes.toBytes("f"),Bytes.toBytes("value_level"));


                Result result = table.get(get);
                byte[] ageBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("age"));
                byte[] genderBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("gender"));
                byte[] valueLevelBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("value_level"));

                int age = Integer.parseInt(Bytes.toString(ageBytes));
                String gender = Bytes.toString(genderBytes);
                String valueLevel = Bytes.toString(valueLevelBytes);

                // 判断该用户的静态画像条件是否满足
                if(age>25 && gender.equals("male") && valueLevel.equals("C")) {

                    message.put("urser_id", userAction.getUser_id());
                    message.put("hit_time", userAction.getAction_time());

                    out.collect(message.toJSONString());
                }
            }
        }
    }

    @Override
    public String ruleId() {
        return this.ruleId;
    }
}
